package org.zoomdev.zoom;

import org.zoomdev.zoom.dao.datasource.RawDataSource;

import javax.sql.DataSource;
import java.time.Duration;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class MigrationExample {

    public static void main(String[] args) {
        // 初始化数据源（实际项目中从连接池获取）
        DataSource dataSource = initializeDataSource();

        // 创建迁移服务
        OnlineDataMigrationService.MigrationService migrationService = new OnlineDataMigrationService.MigrationService(dataSource);

        // 配置迁移
        OnlineDataMigrationService.MigrationConfig config = new OnlineDataMigrationService.MigrationConfig();
        config.setBatchSize(1000);
        config.setDualWritePeriod(Duration.ofHours(2));

        // 指定要迁移的商户和表
        Set<String> merchantIds = new HashSet<>(Arrays.asList("MER001", "MER002", "MER005"));
        String sourceTable = "orders_hot";
        String targetTable = "orders_2023_bak";

        // 启动迁移
        String migrationId = migrationService.startMigration(
                merchantIds, sourceTable, targetTable, config
        );

        // 监听迁移进度
        ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();
        monitor.scheduleAtFixedRate(() -> {
            int progress = migrationService.getMigrationProgress(migrationId);
            MigrationState state = migrationService.getMigrationState(migrationId);
            System.out.printf("Migration [%s] state: %s, progress: %d%%%n",
                    migrationId, state, progress);
        }, 0, 30, TimeUnit.SECONDS);

        // 模拟迁移过程中的中断和恢复
        new Thread(() -> {
            try {
                Thread.sleep(120_000);
                System.out.println("Pausing migration...");
                migrationService.pauseMigration(migrationId);
                Thread.sleep(60_000);
                System.out.println("Resuming migration...");
                migrationService.resumeMigration(migrationId);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();

        // 等待迁移完成
        while (migrationService.getMigrationState(migrationId) != MigrationState.COMPLETED &&
                migrationService.getMigrationState(migrationId) != MigrationState.ROLLBACK) {
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

        // 关闭资源
        monitor.shutdown();
        System.out.println("Migration finished with state: " +
                migrationService.getMigrationState(migrationId));
    }

    private static DataSource initializeDataSource() {
        // 实际项目中使用连接池如 HikariCP
        RawDataSource dataSource = new RawDataSource("jdbc:mysql://localhost:3306/order_db",
                "root","root");
        return dataSource;
    }
}